package drds.data_propagate.parse.binlog_event_position_manager;

import drds.data_propagate.entry.position.BinLogEventPosition;
import drds.data_propagate.parse.exception.ParseException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;


public class PeriodMixedBinLogEventPositionManager extends AbstractBinLogEventPositionManager {

    private static final Logger logger = LoggerFactory.getLogger(PeriodMixedBinLogEventPositionManager.class);
    @SuppressWarnings("serial")
    private final BinLogEventPosition nullBinLogEventPosition = new BinLogEventPosition() {
    };
    private MemoryBinLogEventPositionManager memoryLogPositionManager;
    private ZooKeeperBinLogEventPositionManager zooKeeperLogPositionManager;
    private ScheduledExecutorService scheduledExecutorService;
    private long period;
    private Set<String> persistTasks;

    public PeriodMixedBinLogEventPositionManager(MemoryBinLogEventPositionManager memoryLogPositionManager,
                                                 ZooKeeperBinLogEventPositionManager zooKeeperLogPositionManager, long period) {
        if (memoryLogPositionManager == null) {
            throw new NullPointerException("null memoryLogPositionManager");
        }

        if (zooKeeperLogPositionManager == null) {
            throw new NullPointerException("null zooKeeperLogPositionManager");
        }

        if (period <= 0) {
            throw new IllegalArgumentException("period must be positive, given: " + period);
        }

        this.memoryLogPositionManager = memoryLogPositionManager;
        this.zooKeeperLogPositionManager = zooKeeperLogPositionManager;
        this.period = period;
        this.persistTasks = Collections.synchronizedSet(new HashSet<String>());
        this.scheduledExecutorService = Executors.newScheduledThreadPool(1);
    }

    @Override
    public void stop() {
        super.stop();

        if (zooKeeperLogPositionManager.isStart()) {
            zooKeeperLogPositionManager.stop();
        }

        if (memoryLogPositionManager.isStart()) {
            memoryLogPositionManager.stop();
        }

        scheduledExecutorService.shutdown();
    }

    @Override
    public void start() {
        super.start();

        if (!memoryLogPositionManager.isStart()) {
            memoryLogPositionManager.start();
        }

        if (!zooKeeperLogPositionManager.isStart()) {
            zooKeeperLogPositionManager.start();
        }

        // 启动定时工作任务
        scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

            public void run() {
                List<String> destinationList = new ArrayList<String>(persistTasks);
                for (String destination : destinationList) {
                    try {
                        // 定时将内存中的最新值刷到zookeeper中，多次变更只刷一次
                        zooKeeperLogPositionManager.persistBinLogEventPosition(destination, getLatestBinLogEventPosition(destination));
                        persistTasks.remove(destination);
                    } catch (Throwable e) {
                        // ignore
                        logger.error("period update" + destination + " curosr failed!", e);
                    }
                }
            }
        }, period, period, TimeUnit.MILLISECONDS);
    }

    @Override
    public BinLogEventPosition getLatestBinLogEventPosition(String destination) {
        BinLogEventPosition binLogEventPosition = memoryLogPositionManager.getLatestBinLogEventPosition(destination);
        if (binLogEventPosition == nullBinLogEventPosition) {
            return null;
        } else {
            return binLogEventPosition;
        }
    }

    @Override
    public void persistBinLogEventPosition(String destination, BinLogEventPosition binLogEventPosition) throws ParseException {
        persistTasks.add(destination);
        memoryLogPositionManager.persistBinLogEventPosition(destination, binLogEventPosition);
    }
}
